package SparkMLlib.SparkMLlibDemo

import org.apache.log4j.{Level, Logger}
import org.apache.spark.mllib.recommendation.{ALS, MatrixFactorizationModel, Rating}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

/**
  * 基于RDD的ALS
  * 关于车速、天气、车型、道路线性数据与超载的关系模型
  * 1：速度与超载数据关系：overspeed.csv
  * 2：天气与超载数据关系：weather.csv
  * 3：车型与超载数据关系：vehicletype.csv
  *
  * 关于速度与超载模型关系：速度应该按照划档计算，如，0-30为一档，30-50为二档，50-70为三档，70-90为四档，90以上为五档。
  * 则可以将超速模型构建为：0-30（1），30-50（2），50-70（3），70-90（4），90以上（5）
  *
  * 关于天气与超载模型关系：1：晴天，2：晴转多云，3：雨天，4：雾天，5：雪天
  *
  * 关于车型与超载模型关系：1：客车，2：货车，3：危化品车
  *
  */
object ALSCFDemo {
  //屏蔽日志信息
  Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)
  Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)

  /**
    * 解析数据：将数据转换成Rating对象
    *
    * @param str
    * @return
    */
  def parseRating(str: String): Rating = {
    val fields = str.split(",")
    assert(fields.size == 3)
    Rating(fields(0).toInt, fields(1).toInt, fields(2).toFloat)
  }

  /**
    * @param model 训练好的模型
    * @param data  真实数据
    * @param n     数据个数
    * @return 误差
    */
  def rems(model: MatrixFactorizationModel, data: RDD[Rating], n: Long): Double = {
    //预测值 Rating(userId,itermId,rating)
    val preRDD: RDD[Rating] = model.predict(data.map(d => (d.user, d.product)))
    //关联：组成（预测评分，真实评分）
    val doubleRating = preRDD.map(
      x => ((x.user, x.product), x.rating)
    ).join(
      data.map { x => ((x.user, x.product), x.rating) }
    ).values
    //计算RMES
    math.sqrt(doubleRating.map(x => math.pow(x._1 - x._2, 2)).reduce(_ + _) / n)
  }

  def main(args: Array[String]): Unit = {
    //定义切入点
    val spark = SparkSession.builder().master("local").appName("ALSCFDemo").getOrCreate()
    //读取数据，生成RDD并转换成Rating对象
    //ALSCFDemo.csv：第一列为用户id（userId），第二列为物品id（itemId），第三列为用户给物品的评分
    val overSpeedRDD = spark.sparkContext.textFile("D://GitProjects//SparkKafkaHadoopZookeeperHBaseHiveRedis//spark//src//main//resource//overspeed.csv").map(parseRating)
    val weatherRDD = spark.sparkContext.textFile("D://GitProjects//SparkKafkaHadoopZookeeperHBaseHiveRedis//spark//src//main//resource//weather.csv").map(parseRating)
    val vehicleTypeRDD = spark.sparkContext.textFile("D://GitProjects//SparkKafkaHadoopZookeeperHBaseHiveRedis//spark//src//main//resource//vehicletype.csv").map(parseRating)
    //隐藏因子数
    val rank = 50
    //最大迭代次数
    val maxIter = 10
    //正则化因子
    val labmda = 0.01
    //推荐算法训练模型
    val overSpeedModel = ALS.train(overSpeedRDD, rank, maxIter, labmda)
    val weatherModel = ALS.train(weatherRDD, rank, maxIter, labmda)
    val vehicleTypeModel = ALS.train(vehicleTypeRDD, rank, maxIter, labmda)
    //推荐物品数
    val proNum = 3
    //推荐
    val overSpeedResult = overSpeedModel.recommendProductsForUsers(proNum)
    val weatherResult = weatherModel.recommendProductsForUsers(proNum)
    val vehicleTypeResult = vehicleTypeModel.recommendProductsForUsers(proNum)
    //打印车速与超载关系推荐结果
    println("车速与超载关系推荐结果：")
    overSpeedResult.sortBy(-_._1).foreach(x => {
      println("车牌号： " + x._1)
      x._2.foreach(x => {
        println("速度：" + x.product + ", 超载概率 " + x.rating)
      })
    })
    //打印天气与超载关系推荐结果
    println("天气与超载关系推荐结果：")
    weatherResult.sortBy(-_._1).foreach(x => {
      println("车牌号： " + x._1)
      x._2.foreach(x => {
        println("天气：" + x.product + ", 超载概率 " + x.rating)
      })
    })
    //打印车型与超载关系推荐结果
    println("车型与超载关系推荐结果：")
    vehicleTypeResult.sortBy(-_._1).foreach(x => {
      println("车牌号： " + x._1)
      x._2.foreach(x => {
        println("车型：" + x.product + ", 超载概率 " + x.rating)
      })
    })

    val Array(overSpeedTraining, overspeed) = overSpeedRDD.randomSplit(Array(1, 0))
    val Array(weatherTraining, weather) = weatherRDD.randomSplit(Array(1, 0))
    val Array(vehicleTypeTraining, vehicleType) = vehicleTypeRDD.randomSplit(Array(1, 0))
    //Rems误差训练模型
    val overSpeedRemsModel = ALS.train(overSpeedTraining, rank, maxIter, labmda)
    val weatherRemsModel = ALS.train(weatherTraining, rank, maxIter, labmda)
    val vehicleTypeRemsModel = ALS.train(vehicleTypeTraining, rank, maxIter, labmda)
    //计算误差
    val overSpeedRemsValue = rems(overSpeedRemsModel, overSpeedRDD, overSpeedRDD.count)
    println("车速与超载计算关系模型误差：  " + overSpeedRemsValue)
    //计算误差
    val weatherRemsValue = rems(weatherRemsModel, weatherRDD, weatherRDD.count)
    println("天气与超载计算关系模型误差：  " + weatherRemsValue)
    //计算误差
    val vehicleTypeRemsValue = rems(vehicleTypeRemsModel, vehicleTypeRDD, vehicleTypeRDD.count)
    println("车型与超载计算关系模型误差：  " + vehicleTypeRemsValue)

  }
}

